package com.example.kafka.service;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.stereotype.Service;

import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 反例：多线程并发发送，每个线程都新建 Producer 实例
 */
@Service
public class ProducerServiceMultiThreadSendWrong {

    private final ExecutorService executor = Executors.newFixedThreadPool(5);

    public void sendMultiThread(String topic, String message) {
        executor.submit(() -> {
            Properties props = new Properties();
            props.put("bootstrap.servers", "localhost:9092");
            props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
            props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
            KafkaProducer<String, String> producer = new KafkaProducer<>(props);
            try {
                producer.send(new ProducerRecord<>(topic, message));
            } finally {
                producer.close();
            }
        });
    }
}